package com.zhang.spark_2.com.zhang.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 * @title: 热门商品Top10
 * @author: zhang
 * @date: 2022/2/16 17:16 
 */
object Spark01_req_acc {

  def main(args: Array[String]): Unit = {
    //TODO 获取环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    val sc = new SparkContext(conf);
    //注册累计器
    val hotTop = new HotTop()
    sc.register(hotTop, "hot")
    val userVisit: RDD[String] = sc.textFile("data/user_visit_action.txt")

    userVisit.foreach(
      line => {
        val fields: Array[String] = line.split("_")
        if (fields(6) != "-1") {
          hotTop.add((fields(6), "click"))
        } else if (fields(8) != "null") {
          val strs: Array[String] = fields(8).split(",")
          strs.foreach(
            str => {
              hotTop.add((str, "order"))
            }
          )
        } else if (fields(10) != "null") {
          val strs: Array[String] = fields(10).split(",")
          strs.foreach(
            str => {
              hotTop.add((str, "pay"))
            }
          )
        }
      }
    )

    hotTop.value.map(_._2).toList.sortWith(
      (l, r) => {
        if (l.clickCnt > r.clickCnt) {
          true
        } else if (l.clickCnt == r.clickCnt) {
          if (l.orderCnt > r.orderCnt) {
            true
          } else if (l.orderCnt == r.orderCnt) {
            if (l.payCnt > r.payCnt) {
              true
            } else {
              false
            }
          } else {
            false
          }
        } else {
          false
        }
      }
    ).take(10).foreach(println)

    sc.stop()
  }

  case class HotCategory(cid: String, var clickCnt: Int, var orderCnt: Int, var payCnt: Int)

  class HotTop extends AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] {
    val hotMap = mutable.Map[String, HotCategory]()

    override def isZero: Boolean = {
      hotMap.isEmpty
    }

    override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = {
      new HotTop()
    }

    override def reset(): Unit = {
      hotMap.clear()
    }

    override def add(in: (String, String)): Unit = {
      val cid: String = in._1
      val actionType: String = in._2
      val category: HotCategory = hotMap.getOrElse(cid, HotCategory(cid, 0, 0, 0))
      if (actionType == "click") {
        category.clickCnt += 1
      } else if (actionType == "order") {
        category.orderCnt += 1
      } else {
        category.payCnt += 1
      }

      hotMap.update(cid, category)
    }

    override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
      val map2: mutable.Map[String, HotCategory] = other.value
      map2.foldLeft(hotMap) {
        (map, kv) => {
          val cid: String = kv._1
          val category: HotCategory = kv._2
          val hotCategory: HotCategory = map.getOrElse(cid, HotCategory(cid, 0, 0, 0))
          category.clickCnt += hotCategory.clickCnt
          category.orderCnt += hotCategory.orderCnt
          category.payCnt += hotCategory.payCnt
          map.update(cid,category)
          map
        }
      }
    }

    override def value: mutable.Map[String, HotCategory] = {
      hotMap
    }
  }
}
